package com.xfg;
import org.apache.flume.*;
import org.apache.flume.conf.Configurable;
import org.apache.flume.sink.*;
public class MySink extends AbstractSink implements Configurable {
    Email email=new Email();
    @Override
    public Status process()  {
        //声明返回值状态信息
        Status status = Status.READY;
        //获取当前 Sink 绑定的 Channel
        Channel ch = getChannel();
        //获取事务
        Transaction txn = ch.getTransaction();
        //声明事件
        Event event;
        //开启事务
        txn.begin();
        //读取 Channel 中的事件，直到读取到事件结束循环
            event = ch.take();
                try {
                    if (event != null) {
                        //处理事件（打印）
                        Email.sendEmail("yifan_li@gcoreinc.com","YMS 解析文件失败","hi sir，\r\n出问题了："+event.getHeaders()+"解析失败，请尽快解决!");
                        //事务提交
                        txn.commit();
                        status = Status.READY;
                    }

                } catch (Exception e) {
                    //遇到异常，事务回滚
                    txn.rollback();
                    status = Status.BACKOFF;
                } finally {
                    //关闭事务
                    txn.close();
                }   

        
        
        return status;
    }



    @Override
    public void configure(Context context) {

    }
}
